package org.exhaust.framework.net.server.udp;

import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.*;
import io.netty.channel.socket.DatagramChannel;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.exhaust.framework.net.common.protocol.ChannelEventDispatcher;
import org.exhaust.framework.net.common.protocol.ChannelEventHandler;
import org.exhaust.framework.net.common.protocol.EventLoopFactory;
import org.exhaust.framework.net.common.protocol.RunningState;
import org.exhaust.framework.net.common.protocol.decoder.udp.UdpDecoder;
import org.exhaust.framework.net.common.protocol.encoder.UdpFrameEncoder;
import org.exhaust.framework.net.common.protocol.handler.ConnectionMonitorHandler;

import java.util.function.Supplier;

@Slf4j
public class BaseUdpServer implements UdpServer {

    private volatile Channel channel;

    private volatile RunningState state = RunningState.TERMINATED;

    private ChannelEventHandler handler;

    private Bootstrap bootstrap;

    private EventLoopGroup eventLoopGroup;

    private Supplier<UdpDecoder> decoderSupplier;

    private UdpFrameEncoder encoder;

    private UdpServerProperty serverProperty;

    private ConnectionMonitorHandler connectionMonitorHandler;

    /**
     * @param decoderSupplier supply中必须每次都new一个新对象，因为ByteToMessageDecoder不是线程安全的
     * @param encoder
     * @param handler
     * @param serverProperty
     */
    public BaseUdpServer(Supplier<UdpDecoder> decoderSupplier, UdpFrameEncoder encoder, ChannelEventDispatcher handler, UdpServerProperty serverProperty) {
        this.decoderSupplier = decoderSupplier;
        this.encoder = encoder;
        this.handler = handler;
        this.serverProperty = serverProperty;
        this.bootstrap = new Bootstrap();
        this.eventLoopGroup = EventLoopFactory.eventLoopGroup(serverProperty.getIoThreadNum(), "Server-Worker-EventLoopGroup");
        this.connectionMonitorHandler = new ConnectionMonitorHandler(handler);
    }

    @Override
    public void start() throws Exception {
        this.state = RunningState.RUNNING;
        String localAddress = StringUtils.defaultString(this.serverProperty.getLocalAddress(), "0.0.0.0");
        log.info("启动UDP客户端监听,本地IP：{}，端口号：{}", localAddress, this.serverProperty.getPort());
        this.bootstrap.group(this.eventLoopGroup)
                .channel(EventLoopFactory.datagramChannelClass())
                .option(ChannelOption.SO_BROADCAST, true)
                .option(ChannelOption.SO_RCVBUF, 2048 * 1024)
                .option(ChannelOption.SO_SNDBUF, 1024 * 1024)
                .option(ChannelOption.SO_REUSEADDR, true)
                .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
                .handler(new ChannelInitializer<DatagramChannel>() {
                    @Override
                    protected void initChannel(DatagramChannel ch) throws Exception {
                        ChannelPipeline pipeline = ch.pipeline();
                        pipeline.addLast("ConnectionMonitorHandler", connectionMonitorHandler);
                        if (decoderSupplier != null) {
                            final UdpDecoder decoder = decoderSupplier.get();
                            if (decoder != null) {
                                pipeline.addLast("FrameDecoderAndHandler", decoder);
                            }
                        }
                        if (encoder != null) {
                            pipeline.addLast("FrameEncoder", encoder);
                        }
                    }
                });

        ChannelFuture f = bootstrap.bind(localAddress, this.serverProperty.getPort()).sync();
        this.channel = f.channel();
        log.info("启动UDP服务器成功");
    }

    @Override
    public RunningState getState() {
        return state;
    }

    @Override
    public void shutdownGracefully() {
        this.state = RunningState.STOPPING;
        if (this.channel != null) {
            try {
                this.channel.close();
            } catch (Throwable e) {
                log.warn("close channel failed:{}", e.getMessage());
            }
        }
        this.channel = null;
        this.eventLoopGroup.shutdownGracefully();
        this.eventLoopGroup = null;
        this.bootstrap = null;
        this.state = RunningState.TERMINATED;
    }

    @Override
    public Channel getChannel() {
        return channel;
    }

    @Override
    public EventLoopGroup getEventLoopGroup() {
        return eventLoopGroup;
    }
}
